package com.example.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Created with IntelliJ IDEA.
 * ClassName: DistributedLock
 * Package: com.example.lock
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-30
 * Time: 11:20
 */

//分布式锁
public class DistributedLock {

    private int sessionTimeout = 2000;
    private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private final ZooKeeper zk;

    private String waitPath;

    private CountDownLatch countDownLatch = new CountDownLatch(1);
    //监听等待
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String currenMode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        //获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                //如果连接上 countDownLatch 释放
                if (watchedEvent.getState() == Event.KeeperState.SaslAuthenticated){
                    countDownLatch.countDown();
                }

                //waitLatch 需要释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }

            }
        });
        //等待ZK正常连接后 往下走程序 提高代码健壮性
        countDownLatch.await();

        //判断根节点locks是否存在
        Stat exists = zk.exists("/locks", false);
        if (exists == null){
            //创建根节点 跟节点 创建永久的
            zk.create("/locks","locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }
    }



    //对zk加锁
    public void zkLock() throws InterruptedException, KeeperException {
        //创建对应的临时带序号的节点
        currenMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        //判断是不是最小的节点 是最小的那么获取到锁 如果不是 那么监听前一个节点

        List<String> children = zk.getChildren("/locks", false);

        //如果children只有一个值 那就直接获取锁
        if(children.size() == 1){
            return;
        }else {
            //排序
            Collections.sort(children);

            //获取节点名称 seq-00000000
            String substring = currenMode.substring("/locks/".length());
            //通过seq-00000000获取节点在children的位置
            int index = children.indexOf(substring);

            //判断
            if (index == -1){
                System.out.println("数据异常");
            }else if (index == 0){
                //就一个节点 获取到锁
                return;
            }else {

                //需要监听 他前一个节点的变化
                //-1 就是前一个节点了
                waitPath = "/locks/" + children.get(index-1);
                zk.getData(waitPath,true,null);
                //等待监听
                waitLatch.await();
                return;

            }

        }

    }

    //解锁
    public void unZkLock() throws InterruptedException, KeeperException {

        //删除节点
        zk.delete(currenMode,-1);

    }

}
